package com.wut.clock.event;

import com.alibaba.fastjson.JSONObject;
import com.wut.clock.common.constant.ClockConstant;
import com.wut.clock.common.util.RedisKeyUtil;
import com.wut.clock.entity.Message;
import com.wut.clock.mapper.MessageMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author ZYH
 * @unit D9Lab
 * @description 评论事件消费者
 * @date 2022/6/10 16:13
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class EventConsumer {

    private final MessageMapper messageMapper;

    private final RedisTemplate redisTemplate;

    @KafkaListener(topics = {ClockConstant.COMMENT_TOPIC,ClockConstant.LIKE_TOPIC,ClockConstant.FOLLOW_TOPIC,ClockConstant.COLLECTION_TOPIC})
    public void handleCommentMessage(ConsumerRecord record){

        if (record==null||record.value()==null){
            log.error("消息的内容为空!");
            return;
        }

        Message message = JSONObject.parseObject(record.value().toString(), Message.class);
        if (message==null){
            log.error("消息格式错误!");
            return;
        }

        messageMapper.insert(message);

    }

    @KafkaListener(topics = {ClockConstant.ARTICLE_VIEW_TOPIC})
    public void handleArticleViewIncrement(ConsumerRecord record) {

        if (record == null || record.value() == null){
            log.error("消息的内容为空!");
            return;
        }

        Map map = JSONObject.parseObject(record.value().toString(), Map.class);
        String ip = map.get("ip").toString();
        Integer articleId = Integer.parseInt(map.get("articleId").toString());

        String key = RedisKeyUtil.getArticleViewsCountKey(articleId);
        Long status = redisTemplate.opsForHyperLogLog().add(key, ip);
        if (status == 0) {
            log.info("ip地址：{}，重复访问文章id：{}",ip,articleId);
        }
    }

}
